package day07;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import po.UserBehavior;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Properties;

/**
 * @Description: 读取kafka数据
 * @Author: ZYX
 * @Date: 2022/2/18 10:17
 * @Version: 1.0
 */
public class Demo02 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "20.18.27.177:9092");
        properties.setProperty("group.id", "consumer-group");
        properties.setProperty("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("auto.offset.reset", "latest");

        env.addSource(new FlinkKafkaConsumer<String>(
                "user-behavior-1",
                new SimpleStringSchema(),
                properties
        ))
                .map(new MapFunction<String, UserBehavior>() {
                    @Override
                    public UserBehavior map(String s) throws Exception {
                        String[] split = s.split(",");
                        return  new UserBehavior(split[0],split[1],split[2],split[3],Long.parseLong(split[4])*1000L );
                    }
                })
                .filter(obj -> "pv".equalsIgnoreCase(obj.behavior))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<UserBehavior>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                            @Override
                            public long extractTimestamp(UserBehavior userBehavior, long l) {
                                return userBehavior.timestamp;
                            }
                        })
                )
                // 放入同一流中
                .keyBy(obj -> true)
                .window(SlidingEventTimeWindows.of(Time.hours(1),Time.minutes(5)))
                .process(new ProcessWindowFunction<UserBehavior, String, Boolean, TimeWindow>() {
                    @Override
                    public void process(Boolean aBoolean, Context context, Iterable<UserBehavior> iterable, Collector<String> out) throws Exception {
                        HashMap<String, Long> hashMap = new HashMap<>();
                        for (UserBehavior userBehavior : iterable) {
                            if (hashMap.containsKey(userBehavior.itemId)){
                                hashMap.put(userBehavior.itemId,hashMap.get(userBehavior.itemId)+1L);
                            }else {
                                hashMap.put(userBehavior.itemId,1L);
                            }
                        }

                        ArrayList<Tuple2<String, Long>> list = new ArrayList<>();
                        for (String key : hashMap.keySet()) {
                            list.add(Tuple2.of(key,hashMap.get(key)));
                        }

                        list.sort(new Comparator<Tuple2<String, Long>>() {
                            @Override
                            public int compare(Tuple2<String, Long> o1, Tuple2<String, Long> o2) {
                                return o2.f1.intValue()-o1.f1.intValue();
                            }
                        });

                        StringBuilder result = new StringBuilder();
                        result.append("--------------------------------")
                                .append("窗口："+new Timestamp(context.window().getStart())+"~~"+new Timestamp(context.window().getEnd()))
                                .append("\n");

                        for (int i = 0; i < 3; i++) {
                            Tuple2<String, Long> element = list.get(i);
                            result.append("第"+(i+1)+"名的商品id："+element.f0+"浏览次数："+element.f1+"\n");
                        }

                        out.collect(result.toString());

                    }
                }).print();

        env.execute();
    }

}
